package cn.doitedu.demo3;

import cn.doitedu.demo1.UserEvent_1;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/20
 * @Desc: 学大数据，上多易教育
 * 相对demo2：规则进一步复杂化
 * 规则： 年龄>25岁,且 性别 = male 的用户，规则生效后，做过 C行为3次以上，当他发生 K 行为，则命中规则
 **/
public class Demo3 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 配置一些必选参数
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");


        // 构建一个kafka source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setTopics("user-events-log")
                .setBootstrapServers("doitedu:9092")
                .setGroupId("g001")
                //.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 将source添加到env，得到数据流
        DataStreamSource<String> jsonStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s1");

        // json解析，封装成javaBean
        SingleOutputStreamOperator<UserEvent_1> beanStream = jsonStream.map(value -> JSON.parseObject(value, UserEvent_1.class));

        // 将数据keyBy(用户)
        KeyedStream<UserEvent_1, Long> keyByStream = beanStream.keyBy(UserEvent_1::getUid);

        // 处理逻辑
        SingleOutputStreamOperator<String> hitMessageStream = keyByStream.process(
                new KeyedProcessFunction<Long, UserEvent_1, String>() {

            final JSONObject jsonObject = new JSONObject();
            Table userProfileTable;
            ValueState<Integer> cntState;


            @Override
            public void open(Configuration parameters) throws Exception {

                org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfiguration.create();
                hbaseConf.set("hbase.zookeeper.quorum", "doitedu:2181");

                Connection connection = ConnectionFactory.createConnection(hbaseConf);
                userProfileTable = connection.getTable(TableName.valueOf("user_profile"));


                // 获取一个用于统计C事件行为次数的valueState
                cntState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("c_cnt", Integer.class));



            }

            @Override
            public void processElement(UserEvent_1 bean, KeyedProcessFunction<Long, UserEvent_1, String>.Context ctx, Collector<String> out) throws Exception {

                // 动态画像统计
                if (bean.getEvent_id().equals("C")) {
                    cntState.update(cntState.value() == null ? 1 : cntState.value() + 1);
                }


                // 判断用户的本行为，是否是本规则要求的行为：X
                if (bean.getEvent_id().equals("K")) {


                    // 先判断动态画像条件是否满足规则要求
                    if(cntState.value()!=null && cntState.value() >= 3 ) {

                        // 再查询该用户规则所要求的静态画像标签
                        Get get = new Get(Bytes.toBytes(bean.getUid()));
                        get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("age"));
                        get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("gender"));

                        Result result = userProfileTable.get(get);

                        byte[] ageBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("age"));
                        String ageStr = Bytes.toString(ageBytes);
                        int age = Integer.parseInt(ageStr);

                        byte[] genderBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("gender"));
                        String gender = Bytes.toString(genderBytes);


                        // 判断用户的画像标签是否满足规则的要求
                        if (age > 25 && "male".equals(gender)) {
                            // 如果满足规则要求，则推送规则命中的结果消息
                            jsonObject.put("user_id", bean.getUid());
                            jsonObject.put("rule_id", "rule-001");
                            jsonObject.put("hit_time", bean.getTimestamp());


                            out.collect(jsonObject.toJSONString());
                        }
                    }

                }
            }
        });


        // 将命中结果消息，写入kafka
        hitMessageStream.print();

        // 提交执行
        env.execute();

    }
}
